package com.foreveross.fetl.application.impl;

import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.commons.beanutils.BeanUtils;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.springframework.stereotype.Service;

import com.dayatang.querychannel.support.Page;
import com.foreveross.fetl.application.IHbasePageApplication;
import com.foreveross.fetl.common.HBaseTable;
import com.foreveross.fetl.common.hbase.client.HBaseHelper;
import com.foreveross.fetl.common.vo.FetlErrorMessage;
import com.foreveross.fetl.common.vo.model.SinglePlaneInfo;
import com.foreveross.fetl.common.vo.model.doub.DoublePlaneInfo;
import com.foreveross.fetl.util.DateUtils;
import com.foreveross.fetl.util.FetlErrorUtils;
import com.foreveross.fetl.util.HbaseToPlaneInfoUtil;
import com.foreveross.fetl.util.RowkeyUtils;

@Service("hbasePageApplication")
public class HbasePageApplicationImpl implements IHbasePageApplication{

	@Override
	public Page<FetlErrorMessage> pageQuery(String code, Date start, Date end,
			int page, int pageSize) {
		List<FetlErrorMessage> list =new ArrayList<FetlErrorMessage>();
		StringBuffer startRowkey =new StringBuffer();
		StringBuffer endRowkey =new StringBuffer();
		boolean isHasCode=false;
		if(null!=code && !"".equals(code) ){
			startRowkey.append(code);
			endRowkey.append(code);
			if(null==start && null!=end){
				start=end;
			}
			if(null==end && start!=null ){
				end=start;
			}
			if(null!=start){
				startRowkey.append(DateUtils.format(start, "yyyyMMddHH"));
				endRowkey.append(DateUtils.format(end, "yyyyMMddHH"));
			}
			isHasCode=true;
		}
		startRowkey.append("0");
		endRowkey.append("Z");
		
		ResultScanner rs = null;
		
		int startSize=(page-1)*pageSize;
		int endSize=page*pageSize;
		int i=0;
		try {
			if(isHasCode){
				rs = HBaseHelper.getRecords(HBaseTable.TABLE_FETL_ERROR,
						startRowkey.toString(), endRowkey.toString());
			}else{
				long minStamp;
				long maxStamp=new Date().getTime();
				if(null!=start && null!=end){
					minStamp=start.getTime();
					maxStamp=end.getTime();
				}else{
					Date d =DateUtils.parse(DateUtils.format(new Date(), "yyyyMMdd"), "yyyyMMdd");
					minStamp=d.getTime();
				}
				rs = HBaseHelper.getRecordsByTimestamp(HBaseTable.TABLE_FETL_ERROR,
						minStamp, maxStamp);
			}
			for (Result res : rs) {
				if(i>=startSize && i<endSize){
					Map<String, Map<String, String>> curMap = HBaseHelper
							.transformHBaseRowToMap(res);
					FetlErrorMessage message=FetlErrorUtils.getFormHbase(curMap);
					if(null!=message){
						message.setId(new String(res.getRow()));
						list.add(message);
					}
				}
				i++;
			}
			
		} catch (IOException e) {
			e.printStackTrace();
		}finally{
			if(null!=rs){
				rs.close();
			}
		}
		return new Page<FetlErrorMessage>(page,i,pageSize,list);
	}

	@Override
	public Page<DoublePlaneInfo> pageSingleQuery(String type, String channel,
			Date start, Date end, String fromCity, String toCity,
			String flightDate, String returnDate,int page, int pageSize) {
		
		List<DoublePlaneInfo> list =new ArrayList<DoublePlaneInfo>();
		
		if(null==channel || "".equals(channel))
			return new Page<DoublePlaneInfo>(page,0,pageSize,list);
		
		if(null==start && null!=end){
			start=end;
		}
		if(null==end && start!=null ){
			end=start;
		}
		Calendar cstart=Calendar.getInstance();
		cstart.setTime(start);
		Calendar cend=Calendar.getInstance();
		cend.setTime(end);
		
		ResultScanner rs = null;
		
		int startSize=(page-1)*pageSize;
		int endSize=page*pageSize;
		int i=0;
		String startRow=null;
		String endRow=null;
		do{
			if("single".equals(type)){
				startRow=RowkeyUtils.getRowKey(DateUtils.format(cstart.getTime(), "yyyyMMddHH"), channel, fromCity, toCity,
						flightDate, null, RowkeyUtils.START_ROW_KEY);
				endRow=RowkeyUtils.getRowKey(DateUtils.format(cstart.getTime(), "yyyyMMddHH"), channel, fromCity, toCity,
						flightDate, null, RowkeyUtils.STOP_ROW_KEY);
				try {
					rs=HBaseHelper.getRecords(HBaseTable.BASE_DATA_FLIGHT, startRow, endRow, new String[]{"planeinfo"}, null);
					for (Result res : rs) {
						if(i>=startSize && i<endSize){
							Map<String, Map<String, String>> curMap = HBaseHelper
									.transformHBaseRowToMap(res);
							SinglePlaneInfo singe =null;
							 singe=HbaseToPlaneInfoUtil.hbaseToSinglePlaneInfo(curMap, 1, SinglePlaneInfo.class);
							 singe.setRowkey(new String(res.getRow()));
							if(null!=singe){
								DoublePlaneInfo vo=new DoublePlaneInfo();
								try {
									BeanUtils.copyProperties(vo, singe);
									list.add(vo);
								} catch (IllegalAccessException e) {
									e.printStackTrace();
								} catch (InvocationTargetException e) {
									e.printStackTrace();
								}
							}
						}
						i++;
					}
				} catch (IOException e) {
					e.printStackTrace();
				}finally{
					if(null!=rs)
						rs.close();
				}
			}else{
				
				startRow=RowkeyUtils.getDoubleRowKey(DateUtils.format(cstart.getTime(), "yyyyMMddHH"), channel, fromCity, toCity,
						flightDate,returnDate, null, RowkeyUtils.START_ROW_KEY);
				endRow=RowkeyUtils.getDoubleRowKey(DateUtils.format(cstart.getTime(), "yyyyMMddHH"), channel, fromCity, toCity,
						flightDate,returnDate, null, RowkeyUtils.STOP_ROW_KEY);
				
				try {
					rs=HBaseHelper.getRecords("CRAWL_PLANE_FREIGHT_DOUBLE", startRow, endRow, new String[]{"planeinfo","returnPlaneInfosInfo"}, null);
					for (Result res : rs) {
						if(i>=startSize && i<endSize){
							Map<String, Map<String, String>> curMap = HBaseHelper
									.transformHBaseRowToMap(res);
							DoublePlaneInfo vo =null;
							try {
								vo = HbaseToPlaneInfoUtil.hbaseToPlaneInfo(curMap, 1, DoublePlaneInfo.class);
							} catch (Exception e) {
								e.printStackTrace();
							}
							if(null!=vo){
								vo.setRowkey(new String(res.getRow()));
								list.add(vo);
							}
						}
						i++;
					}
				} catch (IOException e) {
					e.printStackTrace();
				}finally{
					if(null!=rs)
						rs.close();
				}
				
			}
			
			cstart.add(Calendar.HOUR_OF_DAY, 1);
		}while(cstart.before(cend));
		
		return new Page<DoublePlaneInfo>(page,i,pageSize,list);
	}
	
	public Map<String, Map<String, String>>  getByRowkey(String rowkey){
		Map<String, Map<String, String>> curMap =new HashMap<String, Map<String, String>>();
		Result rs = null;
		if(null!=rowkey && !"".equals(rowkey)){
			if(rowkey.length()>=50){
				try {
					rs=HBaseHelper.getOneRecord("CRAWL_PLANE_FREIGHT_DOUBLE", rowkey);
					curMap = HBaseHelper.transformHBaseRowToMap(rs);
				} catch (IOException e) {
					e.printStackTrace();
				}
			}else{
				try {
					rs=HBaseHelper.getOneRecord(HBaseTable.BASE_DATA_FLIGHT, rowkey);
					curMap = HBaseHelper.transformHBaseRowToMap(rs);
				} catch (IOException e) {
					e.printStackTrace();
				}
			}
		}
		
		return curMap;
	}

	@Override
	public FetlErrorMessage getById(String rowkey) {
		Result rs = null;
		if(null!=rowkey && !"".equals(rowkey)){
			try {
				rs=HBaseHelper.getOneRecord(HBaseTable.TABLE_FETL_ERROR, rowkey);
				Map<String, Map<String, String>> curMap = HBaseHelper.transformHBaseRowToMap(rs);
				FetlErrorMessage message=FetlErrorUtils.getFormHbase(curMap);
				if(null!=message){
					message.setId(new String(rs.getRow()));
					return message;
				}
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
		return null;
	}
	

}
